package test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author zcc
 * @date 2021/09/06/15:44
 */
public class PoolDemo {

    public static void main(String[] args) {
        ExcoterPool e = new ExcoterPool(100, 15, 10, 10);
        for (int i = 0; i < 100; i++) {
            final int s = i;
            e.submit(new Thread(() -> {
                System.out.println(Thread.currentThread().getName() +"  i="+s+ " :exect poolSize+"+e.getRunSize() + " queue:" + e.queueSize());
                try {
                    Thread.sleep(200);
                } catch (InterruptedException interruptedException) {
                    interruptedException.printStackTrace();
                }
            }));
        }
        try {
            Thread.sleep(800);
        } catch (InterruptedException interruptedException) {
            interruptedException.printStackTrace();
        }
        e.distoryNow();
    }

}

class ExcoterPool {
    public volatile int size;
    private AtomicBoolean stat = new AtomicBoolean(true);
    private int max;
    private int core;
    private int timeout = 0;
    private WorkerQueue<Runnable> queue;
    private List<Thread> taskList = new ArrayList<>();

    //ctl 算法 -> 状态和工作线程数
    /**
     * 为什么是32 - 3呢？
     * 反推另一个问题为什么使用一个ctl(int)来记录工作线程数和整个线程池的生命周期
     * 解：使用二进制来记录工作线程数和整个线程池的生命周期。设计的是 最高三位记录生命周期,底29位记录运行工作线程数
     * 即：aaabbbbbbbbbbbbbbbbbbbbbbbbbbbbb
     *  - a代表线程池的状态
     *  - b代表工作线程数
     *  CAPACITY 的二进制   00011111111111111111111111111111
     *  RUNNING 的二进制    11100000000000000000000000000000
     *  SHUTDOWN 的二进制   00000000000000000000000000000000
     *  STOP 的二进制       00100000000000000000000000000000
    */
    final int BIT = Integer.SIZE - 3;
    final int CAPACITY = (1 << BIT )- 1;
    final int RUNNING = -1 << BIT;
    final int SHUTDOWN = 0 << BIT;
    final int STOP = 1 << BIT;
    private AtomicInteger ctl = new AtomicInteger(ctlPackageOf(RUNNING,0));

    /**
     * c            11100000000000000000000000000001
     * CAPACITY  &  00011111111111111111111111111111
     * ans          00000000000000000000000000000001
     * get运行线程个数
     * @param c
     * @return
     */
    public int workCountOf(int c) {
        return c & CAPACITY;
    }
    /**
     * c            01100000000000000000000000000001
     * CAPACITY   & 11100000000000000000000000000000
     * ans          01100000000000000000000000000000
     * get线程池状态
     * @param c
     * @return
     */
    public int statOf(int c) {return c & ~CAPACITY;}

    /**
     * c            01100000000000000000000000000000
     * CAPACITY  |  00000000000000000000000000000011
     * ans          01100000000000000000000000000011
     * 封装新的ctl
     * @param r
     * @param c
     * @return
     */
    public int ctlPackageOf(int r,int c) {return r | c;}

    public void workCountIncrement(){
        ctl.incrementAndGet();
    }
    public void workCountDecrement(){
        ctl.decrementAndGet();
    }
    public int queueSize(){
        return queue.size();
    }
    public ExcoterPool(int size, int max, int core, int timeout) {
        this.size = size;
        this.max = max;
        this.core = core;
        this.timeout = timeout;
        // 初始化队列
        queue = new WorkerQueue<>(size);
    }
    public int getRunSize(){
        return workCountOf(ctl.get());
    }
    public void distoryNow(){
        ctlPackageOf(STOP,workCountOf(ctl.get()));
        taskList.stream().forEach(r->{
            System.out.println("threadName = "+ r.getName()+" stat=" + r.getState());
            if (r.getState() == Thread.State.BLOCKED||r.getState() == Thread.State.WAITING||r.getState() == Thread.State.TIMED_WAITING){
                r.interrupt();
            }
        });
    }
    public void distory(){
        while (true) {
            if (ctl.get() >= SHUTDOWN || ctl.compareAndSet(ctl.get(),ctlPackageOf(SHUTDOWN,workCountOf(ctl.get()))))break;
        }
    }
    private void addWorker(Runnable r) {
        WorkerTask workerTask = new WorkerTask(r);
        workerTask.start();
        taskList.add(workerTask);
        workCountIncrement();
    }

    public void submit(Runnable r) {
        System.out.println(toBinaryString(ctl.get()));
        // 小于核心线程数直接创建线程直到达到核心线程数
        int c = workCountOf(ctl.get());
        if (c < this.core) {
            addWorker(r);
            return;
        }else if(this.size == queue.size()) {
            addWorker(r);
            return;
        }
        queue.offer(r);
    }

    private class WorkerTask extends Thread implements Runnable {
        private Runnable firstTask;
        public WorkerTask(Runnable firstTask){
            this.firstTask = firstTask;
        }
        @Override
        public void run() {
            Runnable task = null;
            // 如果存在第一个任务
            task = firstTask;
            firstTask = null;
            try {
                while ( task != null || (task = getTask()) != null) {
                    // 执行任务
                    task.run();
                    task = null;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                // 运行线程数减一
                workCountDecrement();
                System.out.println("close worker ans size="+workCountOf(ctl.get()));
            }

        }

        public Runnable getTask() throws Exception {
            // 检测运行状态：safe失败-> 存在任务执行完再关闭
            boolean timer = workCountOf(ctl.get())> core || !getRunStat();
            // 未超过核心线程数使用take(),超过核心线程使用poll(time)
            return timer ? queue.poll(timeout,TimeUnit.SECONDS):queue.take();
        }

        public Boolean getRunStat() {
            return statOf(ctl.get()) < SHUTDOWN;
        }
    }

    private class WorkerQueue<T> extends LinkedBlockingQueue<T> {
        public WorkerQueue(int size) {
            super(size);
        }

        @Override
        public boolean offer(T t) {
            return super.offer(t);
        }

        @Override
        public T poll(long timeout, TimeUnit unit) throws InterruptedException {
            return super.poll(timeout, unit);
        }
    }
    // 替代Integer.toBinaryString()
    public static String toBinaryString(int num) {
        if (num == 0) return ""+0;
        String result = "";
        // 左面0的个数
        int n = Integer.numberOfLeadingZeros(num);
        num <<= n;
        for (int i=0; i<32-n; ++i) {
            int x = (Integer.numberOfLeadingZeros(num) == 0)?1:0;
            result += x;
            num <<= 1;
        }
        return result;
    }
}
